package org.example.data_stream;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * @author shenguangyang
 */
public class E04_TransformMap {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        List<Event> events = new ArrayList<Event>() {{
            add(new Event("user02", "./cart", 1660095536L));
            add(new Event("user01", "./home", 1660095536L));
            add(new Event("user01", "./home", 1660095536L));
        }};
        DataStream<Event> streamSource = env.fromCollection(events);
        // 进行转换, 提取user字段
        DataStream<String> stream1 = streamSource.map(new MyMapper());
        DataStream<String> stream2 = streamSource.map((Event::getUser));

        stream1.print("stream1");
        stream2.print("stream2");

        env.execute();
    }

    // 自定义 MapFunction
    public static class MyMapper implements MapFunction<Event, String> {
        @Override
        public String map(Event event) throws Exception {
            return event.getUser();
        }
    }
}
